package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator

object Demo20Accumulator {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("map算子演示")
    val context = new SparkContext(conf)
    //====================================================

    val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")

    val scoreRDD: RDD[String] = context.textFile("spark/data/score.txt")

//    var count = 0
//    studentRDD.foreach((line:String)=>{
//      count+=1
//      println("-------------------------")
//      println(count)
//      println("-------------------------")
//    })
//    println(s"count的值为：${count}")

    /**
     * 累加器
     *
     * 由SparkContext来创建
     * 注意：
     *  1、因为累加器的执行实在RDD中执行的，而RDD是在Executor中执行的，而要想在Executor中执行就得有一个action算子触发任务调度
     *  2、sparkRDD中无法使用其他的RDD
     *  3、SparkContext无法在RDD内部使用，因为SparkContext对象无法进行序列化，不能够通过网络发送到Executor中
     */

//    val accumulator: LongAccumulator = context.longAccumulator
//    studentRDD.foreach((line:String)=>{
//      accumulator.add(1)
//    })

//    studentRDD.map((line:String)=>{
//      accumulator.add(1)
//    }).collect()
//    println(s"accumulator的值为：${accumulator.value}")

//    val value: RDD[RDD[(String, String)]] = studentRDD.map((stuLine: String) => {
//      scoreRDD.map((scoreLine: String) => {
//        val strings: Array[String] = scoreLine.split(",")
//        val strings1: Array[String] = stuLine.split(",")
//        val str1: String = strings.mkString("|")
//        val str2: String = strings1.mkString("|")
//        (str1, str2)
//      })
//    })

//    value.foreach(println)

//    val value: RDD[RDD[String]] = studentRDD.map((stuLine: String) => {
//      val scoreRDD: RDD[String] = context.textFile("spark/data/score.txt")
//      scoreRDD
//    })
//    value.foreach(println)




  }
}
